package com.mm.base.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Description Spark大数据去重
 * @Date 2023/11/10 下午1:46
 * @Author yanglin
 **/
@Slf4j
@Configuration
public class SparkConfig {

    /**
     * SparkContext：从Spark1.x开始，Spark SparkContext是Spark的入口点，用于在集群上以编程方式创建Spark RDD、
     * 累加器和广播变量。是spark执行环境的客户端，是spark执行作业的入口点，是spark应用程序的主控。
     *
     * SparkSession：从Spark2.0开始，SparkSession已经成为Spark处理RDD、DataFrame 和 Dataset 的入口点。
     * SparkSession封装了SparkConf、SparkContext和SQLContext。为了向后兼容，SQLContext和HiveContext也被保存下来。
     * 它实质上是SQLContext和HiveContext的组合（未来可能还会加上StreamingContext），所以在SQLContext和HiveContext上可用的API在
     * SparkSession 上同样是可以使用的。SparkSession 内部封装了sparkContext，所以计算实际上是由 sparkContext 完成的。
     *
     * SQLContext：在Spark version1.0中，为了处理结构化数据(行和列)，SQLContext (org.apache.spark.sql.SQLContext )
     * 是一个入口点，但是在2.0版本中，SQLContext已经被SparkSession所取代。Apache Spark SQLContext是SparkSQL的入口点，
     * Spark是Spark1.x中用于结构化数据(行和列)的Spark模块。正在处理。Spark SQLContext已初始化。
     *
     * JavaSparkContext 是 Java友好版本的[org.apache.spark.SparkContext]返回[org.apache.spark.api.java.JavaRDD]，
     * 并使用Java集合，而不是Scala集合。
     */

    @Value("${spark.app.name}")
    private String appName;
    @Value("${spark.master.uri}")
    private String sparkMasterUri;

    /**
     * Spark 基础信息配置
     *
     * @return
     */
    @Bean
    public SparkConf sparkConf() {
        SparkConf sparkConf = new SparkConf()
                .setAppName(appName)
                .setMaster(sparkMasterUri);
        return sparkConf;
    }

    /**
     * 基于 sparkConf 生成
     *
     * @return
     */
    @Bean
    @ConditionalOnMissingBean(JavaSparkContext.class)
    public JavaSparkContext javaSparkContext() {
        return new JavaSparkContext(sparkConf());
    }

    /**
     * 基于 SparkContext 生成
     *
     * @return
     */
    @Bean
    public SparkSession sparkSession() {
        return SparkSession
                .builder()
                .sparkContext(javaSparkContext().sc())
                .getOrCreate();
    }
}
